package com.hrt.kuducode;

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;

import java.util.Arrays;

/**
 * 使用Java api 操作Kudu
 *  1.创建表
 *   1).创建KuduClient
 *   2).创建表的Schema
 *   3).创建CreateTableOptions设置分区规则
 *   4).创建表
 *   5).关闭KuduClient.close()
 *  2.插入、更新、删除数据
 *     1) 创建 KuduClient
 *     2) 创建KuduSession
 *     3) 创建插入、更新、删除对象,组织数据
 *     4).session.apply()
 *     5).关闭 session、client对象
 */
public class JavaKuduApi {
    public static void main(String[] args) throws KuduException {
        //创建表
        CreateTable();

        //插入数据
//        InsertData();

        //查询数据
        ScannerData();

        //修改数据
//        UpdateData();

        //删除数据
//        DeleteData();

//        KuduClient client = new KuduClient.KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build();
//        if(client.tableExists("kudu_t1")){
//            client.deleteTable("kudu_t1");
//        }

//        ScannerData();


    }

    /**
     * 删除数据
     * @throws KuduException
     */
    private static void DeleteData() throws KuduException {
        KuduClient client = new KuduClient.KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build();
        if(client.tableExists("kudu_t1")){
            KuduTable kudu_t1 = client.openTable("kudu_t1");

            KuduSession session = client.newSession();

            Delete delete = kudu_t1.newDelete();
            PartialRow row = delete.getRow();
            row.addInt("id", 10);
            session.apply(delete);

        }
    }

    /**
     * 更新数据
     * @throws KuduException
     */
    private static void UpdateData() throws KuduException {
        KuduClient client = new KuduClient.KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build();
        if(client.tableExists("kudu_t1")){
            KuduTable kudu_t1 = client.openTable("kudu_t1");
            KuduSession session = client.newSession();

//            Update update = kudu_t1.newUpdate();
            Upsert update = kudu_t1.newUpsert();
            PartialRow row = update.getRow();
            row.addInt("id",10 );
            row.addString("name","ww" );
            row.addInt("age",20 );
            row.addDouble("score",300 );
            session.apply(update);

        }
    }

    /**
     * 查询数据
     * @throws KuduException
     */
    private static void ScannerData() throws KuduException {
        KuduClient client = new KuduClient.KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build();
        if(client.tableExists("flink_t1")){
            KuduTable kudu_t1 = client.openTable("flink_t1");
            KuduScanner ks = client.newScannerBuilder(kudu_t1).build();
            //是否有多个tablet
            while(ks.hasMoreRows()){
                //获取其中一个tablet
                RowResultIterator iter = ks.nextRows();
                //判断tablet中是否有数据
                while(iter.hasNext()){
                    RowResult next = iter.next();
                    int id = next.getInt("id");
                    String name = next.getString("name");
                    int age = next.getInt("age");
                    double score = next.getDouble("score");
                    System.out.println("id = "+id +",name = "+name+",age= "+age+",score = "+score);
//                    System.out.println("id = "+id +",name = "+name+",age= "+age);
                }
            }
        }
    }

    /**
     * 插入数据
     * @throws KuduException
     */
    private static void InsertData() throws KuduException {
        KuduClient client = new KuduClient.KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build();
        if(client.tableExists("kudu_t1")){
            //创建连接session
            KuduSession session = client.newSession();
            //打开表
            KuduTable tbl = client.openTable("kudu_t1");
            //创建插入数据的操作
            Insert insert = tbl.newInsert();
            PartialRow row = insert.getRow();
            row.addInt("id",3 );
            row.addString("name","zs" );
            row.addInt("age",18 );
            row.addDouble("score",100 );


            session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
            //执行插入操作
            session.apply(insert);
            session.flush();

        }
    }

    /**
     * 创建表
     * @throws KuduException
     */
    private static void CreateTable() throws KuduException {
        //创建KuduClient对象
        KuduClient client = new KuduClient.KuduClientBuilder("cm1:7051,cm2:7051,cm3:7051").build();

        //创建表 kudu_t1
        if(!client.tableExists("flink_t1")){
            //指定表中的各个列信息
            Schema schema = new Schema(Arrays.asList(
                    // .key(true)表示主键
                    new ColumnSchema.ColumnSchemaBuilder("id", Type.INT32).key(true).build(),
                    new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).build(),
                    new ColumnSchema.ColumnSchemaBuilder("age", Type.INT32).build(),
                    new ColumnSchema.ColumnSchemaBuilder("score", Type.DOUBLE).build()
            ));

            //设置表的分区规则 5个分区
            CreateTableOptions options = new CreateTableOptions();
            options.addHashPartitions(Arrays.asList("id"), 5);

            client.createTable("flink_t1", schema, options);
        }

        client.close();
    }
}
